// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: gekim
#pragma once
#include <fmt/format.h>

#include <algorithm>
#include <boost/algorithm/string/predicate.hpp>
#include <ctime>  // strptime
#include <experimental/filesystem>
#include <memory>
#include <regex>
#include <set>
#include <string>
#include <utility>
#include <vector>

#include "cc/coin2/feed/fastfeed/reader.h"
#include "coin/feed/fastfeed/feed.pb.h"
#include "coin2/base/config/app_config.h"
#include "coin2/exchange/base/market/market_exchange_api.h"
#include "coin2/exchange/base/symbology/product.h"
#include "coin2/exchange/factory/symbology/product.h"
#include "coin2/feed/fastfeed/const.h"
#include "orio/io/base.h"
#include "orio/io/concat.h"
#include "orio/record/simple.h"

namespace impl {
using coin2::exchange::symbology::ProductRequest;
using fastfeed::proto::Feed;

// bool cmp(const std::experimental::filesystem::path& lhs,
//          const std::experimental::filesystem::path& rhs) {
//   if (lhs.parent_path() != rhs.parent_path()) {
//     return lhs < rhs;
//   } else {
//     std::string lhs_file_name_core =
//         lhs.filename().string().substr(0, lhs.filename().string().rfind(".W"));
//     std::string rhs_file_name_core =
//         rhs.filename().string().substr(0, rhs.filename().string().rfind(".W"));
//     return lhs_file_name_core < rhs_file_name_core;
//   }
// }

class FastFeedUniverseReader : public orio::io::Reader<Feed> {
 public:
  // /{proto_feed_universe_root}/{machine}/{universe}/{YYMMDD}/a
  // /remote/iosg/data/repository/feed/universe/{machine}/{universe}/{YYMMDD}/b
  // /remote/iosg/data/repository/feed/universe/{machine}/{universe}/{YYMMDD}/c
  //
  // /remote/iosg/data/repository/feed/proto/{machine}/{exchange}/{YYYYMMDD}/{product}/0.day.20200508-153739+0900

  FastFeedUniverseReader(
      const MarketExchangeApiRecipe& mear,
      const ProductConfig& config,
      const std::string& proto_feed_universe_root,
      const std::string& machine,
      const std::set<std::string>& universe_tags,
      int64_t warmup,
      int64_t begin_timestamp, /* inclusive */
      int64_t end_timestamp,   /* exclusive */
      size_t max_record_size) {
    std::vector<std::unique_ptr<orio::io::Reader<Feed>>> readers;

    auto products = ProductRequestsFromConfig(mear.mea.String(), config, begin_timestamp);
    for (auto& prod : products) {
      products_mea_[prod->relative_norm()] = std::move(prod);
    }
    for (auto& universe_tag : universe_tags) {
      LOG(INFO) << universe_tag;
      std::vector<std::experimental::filesystem::path> paths =
          GetPathList(machine, proto_feed_universe_root, universe_tag, mear, begin_timestamp);

      // 20191010/BTC.gz --
      // 20191011/BTC.gz --   -> MultiFileReader -> SimpleReader ->
      // FastFeedProto Reader (max_size)
      // -> TimeFilter (IntervalFastFeedReader)

      // 20191010/ETH.gz --
      // 20191011/ETH.gz --   -> MultiFileReader -> SimpleReader ->
      // FastFeedProto Reader -> IntervalFastFeedReader

      // Merge(BTC, ETH) by timestamp

      if (paths.empty()) {
        LOG(WARNING) << "paths are empty!!";
        continue;
      }
      std::sort(std::begin(paths), std::end(paths), cmp);
      for (auto& path : paths) {
        LOG(INFO) << "[FastFeed Universe Reader] will read " << path;
      }

      readers.push_back(std::unique_ptr<orio::io::Reader<Feed>>(new IntervalFastFeedReader(
          std::unique_ptr<orio::io::Reader<Feed>>(new FastFeedProtoReader(
              std::unique_ptr<orio::record::SimpleMultiFileReader>(
                  orio::record::SimpleMultiFileReader::FromFile(paths, max_record_size)),
              max_record_size)),
          begin_timestamp - warmup,
          end_timestamp)));
    }

    reader_ = std::make_unique<MergedFastFeedReader>(std::move(readers));
  }

  bool Read(Feed* out) override {
    auto symbol_key = products_mea_.find(out->symbol());
    if (symbol_key != products_mea_.end()) {
      products_.insert(std::move(symbol_key->second));
      products_mea_.erase(symbol_key);
    }

    return reader_->Read(out);
  }

  std::vector<IProduct*> products() {
    std::vector<IProduct*> out;
    for (const auto& p : products_) {
      out.push_back(p.get());
    }
    return out;
  }

  // /remote/iosg/coin-2/feed_cache/
  // feed-02.us-east-1.aws/20190608/Spot.Gdax.v1/ETH-BTC.Gdax.ETH-BTC.W2.gz
  // BCHABC-USD.20191122.Okex.BCH1122.W1.gz
  inline static std::string DEFAULT_ROOT = "/remote/iosg/coin-2/feed_cache/";

 private:
  std::vector<std::experimental::filesystem::path> GetPathList(
      const std::string& machine,
      const std::string& root_dir,
      const std::string& universe_tag,
      const MarketExchangeApiRecipe& mear,
      int64_t begin_timestamp) {
    std::vector<std::experimental::filesystem::path> out;

    char yyyymmdd[9];
    time_t t = begin_timestamp / 1'000'000'000ull;
    tm tm_;
    gmtime_r(&t, &tm_);
    std::strftime(yyyymmdd, 9, "%Y%m%d", &tm_);

    auto complete_dir = fmt::format(
        "{}/{}/{}/{}/{}",
        root_dir,
        machine,
        yyyymmdd,
        mear.String(),
        universe_tag + ".W0.lz4");
    out.push_back(complete_dir);
    return out;
  }

  std::unique_ptr<MergedFastFeedReader> reader_;

  // struct _ProductLess {
  //   inline bool operator()(
  //       const std::unique_ptr<IProduct>& lhs,
  //       const std::unique_ptr<IProduct>& rhs) const {
  //     return lhs->key() < rhs->key();
  //   }
  // };

  std::map<std::string, std::unique_ptr<IProduct>> products_mea_;
  std::set<std::unique_ptr<IProduct>> products_;
};

}  // namespace impl

using impl::FastFeedUniverseReader;
